package com.lagou.client;

import com.lagou.pojo.Connect;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.string.StringEncoder;
import org.I0Itec.zkclient.IZkChildListener;
import org.I0Itec.zkclient.ZkClient;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;

import javax.naming.Name;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;

//客户端给服务器发送数据
public class NettyClient implements Watcher {

//    private  static CountDownLatch countDownLatch = new CountDownLatch(1);

    private static final String IP = "127.0.0.1";
    private static final int PORT = 8081;

    private static final String ZOOKEEPER_ADDRESS = "127.0.0.1:2181";

    private static final String NAMESPACE = "dubbo-demo";

    private static ZooKeeper zooKeeper;

    private static List<String> childrens = new ArrayList<>();

    private static List<Connect> connects = new ArrayList<>();


    static ScheduledExecutorService scheduledExecutorService = Executors.newSingleThreadScheduledExecutor();

    static ExecutorService executorService = Executors.newFixedThreadPool(2);


    public static void main(String[] args) throws IOException, InterruptedException {

        zooKeeper = new ZooKeeper(ZOOKEEPER_ADDRESS, 5000, new NettyClient());


//        scheduledExecutorService.scheduleAtFixedRate(NettyClient::updateServerTime, 0, 5, TimeUnit.SECONDS);


        Thread.sleep(Integer.MAX_VALUE);
    }

    @Override
    public void process(WatchedEvent watchedEvent) {
 /*
            子节点列表发生改变时，服务器端会发生noteChildrenChanged事件通知
            要重新获取子节点列表，同时注意：通知是一次性的，需要反复注册监听
         */

        if (watchedEvent.getState() == Event.KeeperState.SyncConnected) {

//            System.out.println("节点发生变化");
            connects.clear();
            try {
                childrens = getChildrens("/");
//                System.out.println("children:" + childrens);
                for (String children : childrens) {
                    if (children.startsWith("dubbo-")) {
                        String substring = children.replace("dubbo-","");
                        System.out.println("noteData>>>>>" + substring);
                        String[] split = substring.split(":");
                        Connect connect = new Connect(split[0], Integer.parseInt(split[1]));
                        executorService.submit(new ConnectThread(connect));
                    }
                }
            } catch (KeeperException e) {
                e.printStackTrace();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }

        }


        // SyncConnected
//        if(watchedEvent.getState() == Event.KeeperState.SyncConnected){
//
//            //解除主程序在CountDownLatch上的等待阻塞
//            System.out.println("process方法执行了...");
//
//            // 获取节点数据的方法
//            try {
//                getNoteData();
//
//                // 获取节点的子节点列表方法
//                getChildrens();
//            } catch (KeeperException e) {
//                e.printStackTrace();
//            } catch (InterruptedException e) {
//                e.printStackTrace();
//            }
//
//
//        }
    }


    /*
       获取某个节点的内容
    */
    private static String getNoteData(String nodeName) throws KeeperException, InterruptedException {

        /**
         * path    : 获取数据的路径
         * watch    : 是否开启监听
         * stat    : 节点状态信息
         *        null: 表示获取最新版本的数据
         *  zk.getData(path, watch, stat);
         */
        byte[] data = zooKeeper.getData("/" + nodeName, false, null);
//        System.out.println("data:" + new String(data));
        return new String(data);

    }

    private static void setNoteData(String nodeName, String data) throws KeeperException, InterruptedException {
        zooKeeper.setData("/" + nodeName, data.getBytes(), -1);

    }


    /*
        获取某个节点的子节点列表方法
     */
    public static List<String> getChildrens(String parent) throws KeeperException, InterruptedException {

        /*
            path:路径
            watch:是否要启动监听，当子节点列表发生变化，会触发监听
            zooKeeper.getChildren(path, watch);
         */
        List<String> children = zooKeeper.getChildren(parent, true);

//        System.out.println("children:" + children);

        return children;

    }

    /**
     * 更新服务端的时间
     */
    public static void updateServerTime() {
        String path = "/" + NAMESPACE + IP + ":" + PORT;
        System.out.println("定时任务开始执行：" + System.currentTimeMillis());
        String noteData = null;
        try {
            noteData = getNoteData(path);
            if (System.currentTimeMillis() - Long.parseLong(noteData) > 5000) {
                setNoteData(path, String.valueOf(System.currentTimeMillis()));
            }
        } catch (KeeperException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}
